package com.pk.flink.source;

import com.alibaba.fastjson.JSONObject;
import com.pk.flink.bean.ChongGouData;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

import java.sql.*;
import java.util.List;

public class RedisChongGouDataSource implements SourceFunction<ChongGouData> {
    private boolean isRunning = true;


    @Override
    public void run(SourceContext<ChongGouData> ctx) throws Exception {
        Jedis jedis = null;
        JedisPool jedisPool = null;

        try {
            JedisPoolConfig config = new JedisPoolConfig();
            config.setMaxTotal(100);
            config.setMaxIdle(10);
            config.setMaxWaitMillis(10000);
            config.setTestOnBorrow(true);
            jedisPool = new JedisPool(config, "ddrankredis1.redis.rds.aliyuncs.com", 6379, 10000,
                    "Ddrank_Redis", 13);

            jedis = jedisPool.getResource();

            List<String> values = jedis.lrange("chonggou_check_shops",0,-1L);

            for(int i=0;i<values.size();i++)
            {
                JSONObject jsonObject = JSONObject.parseObject(values.get(i));
                ChongGouData chongGouData = new ChongGouData(jsonObject.getString("account_guid"),jsonObject.getString("token_prod"));
                ctx.collect(chongGouData);
            }


        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            if (jedisPool != null) {
                jedisPool.close();
                jedisPool = null;
            }

            if (jedis != null) {
                jedis.close();
                jedis = null;
            }
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
